package com.rabbit.rpc;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.UUID;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeoutException;

/**
 * rpc客户端
 */
public class RPCClient {

    // 定义请求队列常量
    private final static String REQUEST_QUEUE_NAME = "rpc_queue";
    // 定义回复队列常量
    private final static String REPLY_QUEUE_NAME = "rpc_reply_queue";

    /**
     * 客户端启动入口
     */
    public static void main(String[] args) {
        System.out.println("Response1: " + call("Hello, RPC Server1"));
        System.out.println("Response2: " + call("Hello, RPC Server2"));
        System.out.println("Response3: " + call("Hello, RPC Server3"));
    }

    /**
     * 发送请求到队列并返回响应数据
     *
     * @param message 请求消息
     * @return 响应数据
     */
    public static String call(String message) {
        // 创建连接
        ConnectionFactory factory = new ConnectionFactory();
        //factory.setHost("localhost");

        String result = "";

        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {

            // 声明请求队列和回复队列
            channel.queueDeclare(REQUEST_QUEUE_NAME, false, false, false, null);
            channel.queueDeclare(REPLY_QUEUE_NAME, false, false, false, null);

            // 每次仅接收一条未经确认的消息
            channel.basicQos(1);

            // 生成关联id
            String correlationId = UUID.randomUUID().toString();

            // 构造请求基本属性
            AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
                    .correlationId(correlationId)   // 设置关联id
                    .replyTo(REPLY_QUEUE_NAME)      // 设置回复队列 (rpc_reply_queue)
                    .build();

            // 发送请求消息到rpc_queue队列
            channel.basicPublish("", REQUEST_QUEUE_NAME, props, message.getBytes(StandardCharsets.UTF_8));

            // 用于保存响应消息的阻塞队列
            final BlockingQueue<String> response = new ArrayBlockingQueue<>(1);

            // 监听回复队列接收响应消息
            String consumerTag = channel.basicConsume(REPLY_QUEUE_NAME, true, new DefaultConsumer(channel) {
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) {
                    // 通过关联id获取到响应数据
                    if (properties.getCorrelationId().equals(correlationId)) {
                        response.offer(new String(body, StandardCharsets.UTF_8));
                    }
                }
            });

            // 清空阻塞队列
            response.clear();

            // 等待接收响应消息
            result = response.take();

            // 取消消费者的监听
            channel.basicCancel(consumerTag);

        } catch (IOException | TimeoutException | InterruptedException e) {
            e.printStackTrace();
        }

        return result;
    }

}